package com.alibaba.otter.canal.instance.core;

import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.ha.CanalHAController;
import com.alibaba.otter.canal.parse.ha.HeartBeatHAController;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.model.Event;

/**
 *
 * AbstractCanalInstance是CanalInstance的抽象子类，
 * 定义了相关字段来维护eventParser、eventSink、eventStore、metaManager的引用。
 * 在AbstractCanalInstance中，并没有提供方法来初始化这些字段。
 * 这些字段都是protected的，子类可以直接访问，显然这些字段都是在AbstractCanalInstance的子类中进行赋值的。
 * 对于instance模块而言，其核心工作逻辑都是在AbstractCanalInstance中实现的。
 *
 * Created with Intellig IDEA. Author: yinxiu Date: 2016-01-07 Time: 22:26
 */
public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {

    private static final Logger                      logger = LoggerFactory.getLogger(AbstractCanalInstance.class);

    protected Long                                   canalId;      // 和manager交互唯一标示                                                // 和manager交互唯一标示
    protected String                                 destination;  // 队列名字                                                // 队列名字
    protected CanalEventStore<Event>                 eventStore;   // 有序队列                                                // 有序队列

    protected CanalEventParser                       eventParser;  // 解析对应的数据信息                                                // 解析对应的数据信息
    protected CanalEventSink<List<CanalEntry.Entry>> eventSink;    // 链接parse和store的桥接器                                                // 链接parse和store的桥接器
    protected CanalMetaManager                       metaManager;  // 消费信息管理器                                                // 消费信息管理器
    protected CanalAlarmHandler                      alarmHandler; // alarm报警机制                                                // alarm报警机制
    protected CanalMQConfig                          mqConfig;                                                     // mq的配置


    // subscribeChange方法，主要是更新一下eventParser中的filter。
    // 关于filter，进行一下补充说明，filter规定了需要订阅哪些库，哪些表。
    // 在服务端和客户端都可以设置，客户端的配置会覆盖服务端的配置。
    // 服务端配置：主要是配置instance.properties中的canal.instance.filter.regex配置项，
    // 官网文档关于这个配置项有以下介绍
    @Override
    public boolean subscribeChange(ClientIdentity identity) {
        if (StringUtils.isNotEmpty(identity.getFilter())) {//如果设置了filter
            logger.info("subscribe filter change to " + identity.getFilter());
            AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());

            boolean isGroup = (eventParser instanceof GroupEventParser);
            if (isGroup) {
                // 处理group的模式
                List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
                for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
                    if(singleEventParser instanceof AbstractEventParser) {
                        ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
                    }
                }
            } else {
                if(eventParser instanceof AbstractEventParser) {
                    ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
                }
            }

        }

        // filter的处理规则
        // a. parser处理数据过滤处理
        // b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份，每份的数据可以根据自己的过滤规则不同而有不同的数据
        // 后续内存版的一对多分发，可以考虑
        return true;
    }
    // 在AbstractCanalInstance的start方法中，主要就是启动各个模块。
    // 启动顺序为：metaManager—>eventStore—>eventSink—>eventParser。
    // 官方关于instance模块构成的图中，把metaManager放在最下面，说明其是最基础的部分，因此应该最先启动。
    // 而eventParser依赖于eventSink，需要把自己解析的binlog交给其加工过滤，
    // 而eventSink又要把处理后的数据交给eventStore进行存储。
    // 因此依赖关系如下：eventStore—>eventSink—>eventParser ，启动的时候也要按照这个顺序启动。
    @Override
    public void start() {
        super.start();
        if (!metaManager.isStart()) {
            metaManager.start();
        }

        if (!alarmHandler.isStart()) {
            alarmHandler.start();
        }

        if (!eventStore.isStart()) {
            eventStore.start();
        }

        if (!eventSink.isStart()) {
            eventSink.start();
        }

        if (!eventParser.isStart()) {
            beforeStartEventParser(eventParser);
            eventParser.start();
            afterStartEventParser(eventParser);
        }
        logger.info("start successful....");
    }
    // 在停止的时候，实际上就是停止内部的各个模块，模块停止的顺序与start方法刚好相反
    // 在AbstractCanalInstance的start和stop方法，对于eventParser这个组件的启动和停止，都有一些特殊处理，
    // 这与eventParser的自身构成有关系。
    // 每个event parser 都会关联两个内部组件： CanalLogPositionManager, CanalHAController
    // CanalLogPositionManager: 记录binlog最后一次解析成功位置信息，描述下一次canal启动的位点。
    //  mysql在主从同步过程中，要求slave自己维护binlog的消费进度信息。canal伪装成slave，因此也要维护这样的信息。
    //  如果自行搭建过mysql主从复制的话，在slave机器的data目录下，都会有一个master.info文件，
    //  这个文件的作用就是存储主库的消费binlog解析进度信息。
    // CanalHAController：控制event parser的链接主机管理，判断当前该链接哪个mysql数据库
    @Override
    public void stop() {
        super.stop();
        logger.info("stop CannalInstance for {}-{} ", new Object[] { canalId, destination });

        if (eventParser.isStart()) {
            beforeStopEventParser(eventParser);
            eventParser.stop();
            afterStopEventParser(eventParser);
        }

        if (eventSink.isStart()) {
            eventSink.stop();
        }

        if (eventStore.isStart()) {
            eventStore.stop();
        }

        if (metaManager.isStart()) {
            metaManager.stop();
        }

        if (alarmHandler.isStart()) {
            alarmHandler.stop();
        }

        logger.info("stop successful....");
    }

    /**
     * eventParser前做的一些特殊处理。
     */
    protected void beforeStartEventParser(CanalEventParser eventParser) {
        //1、判断eventParser的类型是否是GroupEventParser
        boolean isGroup = (eventParser instanceof GroupEventParser);
        if (isGroup) {
            //2、如果是GroupEventParser，则循环启动其内部包含的每一个CanalEventParser，
            // 依次调用startEventParserInternal方法
            // 处理group的模式
            List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
            for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
                startEventParserInternal(singleEventParser, true);
            }
        } else {
            //如果不是，说明是一个普通的CanalEventParser，直接调用startEventParserInternal方法
            startEventParserInternal(eventParser, false);
        }
    }
    // 这个方法内部主要是通过metaManager读取一下历史订阅过这个CanalInstance的客户端信息，然后更新一下filter。
    // around event parser, default impl
    protected void afterStartEventParser(CanalEventParser eventParser) {
        // 读取一下历史订阅的filter信息
        List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
        for (ClientIdentity clientIdentity : clientIdentitys) {
            subscribeChange(clientIdentity);
        }
    }

    // around event parser
    protected void beforeStopEventParser(CanalEventParser eventParser) {
        // noop
    }

    protected void afterStopEventParser(CanalEventParser eventParser) {

        boolean isGroup = (eventParser instanceof GroupEventParser);
        if (isGroup) {
            // 处理group的模式
            List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
            for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
                stopEventParserInternal(singleEventParser);
            }
        } else {
            stopEventParserInternal(eventParser);
        }
    }

    /**
     * 初始化单个eventParser，不需要考虑group
     * isGroup这个参数并没有被用到？？
     */
    protected void startEventParserInternal(CanalEventParser eventParser, boolean isGroup) {
        // 1 、启动CanalLogPositionManager
        if (eventParser instanceof AbstractEventParser) {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (!logPositionManager.isStart()) {
                logPositionManager.start();
            }
        }
        // 2 、启动CanalHAController
        if (eventParser instanceof MysqlEventParser) {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            CanalHAController haController = mysqlEventParser.getHaController();

            if (haController instanceof HeartBeatHAController) {
                ((HeartBeatHAController) haController).setCanalHASwitchable(mysqlEventParser);
            }

            if (!haController.isStart()) {
                haController.start();
            }

        }
    }

    protected void stopEventParserInternal(CanalEventParser eventParser) {
        if (eventParser instanceof AbstractEventParser) {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (logPositionManager.isStart()) {
                logPositionManager.stop();
            }
        }

        if (eventParser instanceof MysqlEventParser) {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            CanalHAController haController = mysqlEventParser.getHaController();
            if (haController.isStart()) {
                haController.stop();
            }
        }
    }

    // ==================getter==================================
    // 通过直接操作容器内部组件来订阅binlog数据
    // AbstractCanalInstance通过相关get方法直接返回了其内部的组件
    // CanalServerWithEmbedded就是直接通过CanalInstance的内部组件，进行操作的。
    // AbstractCanalInstance除了负责启动和停止其内部组件，就没有其他工作了。
    // 真正获取binlog信息，以及相关元数据维护的逻辑，都是在CanalServerWithEmbedded中完成的。
    @Override
    public String getDestination() {
        return destination;
    }

    @Override
    public CanalEventParser getEventParser() {
        return eventParser;
    }

    @Override
    public CanalEventSink getEventSink() {
        return eventSink;
    }

    @Override
    public CanalEventStore getEventStore() {
        return eventStore;
    }

    @Override
    public CanalMetaManager getMetaManager() {
        return metaManager;
    }

    @Override
    public CanalAlarmHandler getAlarmHandler() {
        return alarmHandler;
    }

    @Override
    public CanalMQConfig getMqConfig() {
        return mqConfig;
    }
}
